import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.StatusCodes
import akka.http.scaladsl.server.Directives._
import akka.stream.ActorMaterializer
import slick.jdbc.MySQLProfile.api._
import scala.concurrent.Future
import scala.io.StdIn
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

// 定义数据库表模型
case class ValueAnalysis(value: String, count: Int)

class AnalysisTable(tag: Tag) extends Table[ValueAnalysis](tag, "lc_value_analysis") {
  def value = column[String]("value", O.PrimaryKey)
  def count = column[Int]("count")

  def * = (value, count) <> (ValueAnalysis.tupled, ValueAnalysis.unapply)
}

object SparkRestApi {
  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("sparkRestApiSystem")
    implicit val materializer = ActorMaterializer()
    implicit val executionContext = system.dispatcher

    // 初始化数据库连接
    val db = Database.forConfig("mysqlDB")

    // 实例化数据库表
    val analysisTable = TableQuery[AnalysisTable]

    // 定义 REST API 路由
    val route =
      path("lc" / "value-analysis") {
        get {
          val result: Future[Seq[ValueAnalysis]] = db.run(analysisTable.result)
          onComplete(result) {
            case scala.util.Success(valueAnalysisList) =>
              val serverTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
              val response = Map("data" -> valueAnalysisList, "serverTime" -> serverTime)
              complete(StatusCodes.OK, response.toString())
            case scala.util.Failure(ex) =>
              complete(StatusCodes.InternalServerError, s"An error occurred: ${ex.getMessage}")
          }
        }
      }

    // 启动 HTTP 服务器
    val bindingFuture = Http().newServerAt("localhost", 8080).bind(route)

    println(s"Server online at http://localhost:8080/\nPress RETURN to stop...")
    StdIn.readLine()
    bindingFuture
      .flatMap(_.unbind())
      .onComplete(_ => system.terminate())
  }
}
